/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.reading;

import java.nio.ByteBuffer;

import edu.ucsd.db.bassarisk.data.api.IValueType.IValueReference;
import edu.ucsd.db.bassarisk.data.values.IntType;
import edu.ucsd.db.bassarisk.data.values.IntType.IntTools;
import edu.ucsd.db.bassarisk.data.values.IntType.IntegerValueCodec;
import edu.ucsd.db.bassarisk.storage.common.ColumnStore;
import edu.ucsd.db.bassarisk.storage.common.IColumnSegmentInfo;

public class FixedLengthAscendingIntegerColumnIndexReader implements
    IColumnIndexReader<IntType> {

  private class Cursor {
    protected int drawerIndex;
  }

  private static final int BYTES_IN_HEADER = 20;
  private static final int ENTIRES_IN_FULL_DRAWER = 8064;
  private static final int indexVersionNumber = 2;

  private IColumnSegmentInfo<IntType> columnInfo;
  private final Cursor current;
  private int drawerAddressStart;
  private int firstEntryStart;
  private ByteBuffer headerBuffer;
  private short indexCardCount;
  private IntegerValueCodec intManip;
  private int numIndexedDrawers;
  private long totalValues;
  private int valueIndexStart = 0;

  public FixedLengthAscendingIntegerColumnIndexReader() {
    this.current = new Cursor();
    this.current.drawerIndex = -1; // before the first drawer
  }

  @Override
  public int getColumnDrawerCount() {
    return this.numIndexedDrawers;
  }

  @Override
  public int getColumnIndexVersion() {
    return indexVersionNumber;
  }

  @Override
  public void initialize(final ByteBuffer headerBuffer,
                         final ColumnReader<IntType>.ColumnInfo columnInfo,
                         final int headerCardAddress) {
    this.columnInfo = columnInfo;
    this.headerBuffer = headerBuffer;

    int headerEndOffset = (headerCardAddress + 1) * ColumnStore.BYTES_IN_CARD;

    int headerStartPosition = headerEndOffset - BYTES_IN_HEADER;
    this.headerBuffer.position(headerStartPosition);
    this.totalValues = this.headerBuffer.getLong();
    this.numIndexedDrawers = this.headerBuffer.getInt();
    this.indexCardCount = this.headerBuffer.getShort();
    int maximalValue = this.headerBuffer.getInt();
    assert indexVersionNumber == this.headerBuffer.getShort();

    int spaceRequired = this.indexCardCount * ColumnStore.BYTES_IN_CARD;

    this.firstEntryStart = headerEndOffset - spaceRequired;
    this.drawerAddressStart = this.firstEntryStart + (4 * this.numIndexedDrawers);

    this.intManip = IntTools.INSTANCE.getGenericCodec();
    
    columnInfo.reset(this.totalValues, 
                     new IntType.ObjectIntegerReference(Integer.MIN_VALUE), 
                     new IntType.ObjectIntegerReference(maximalValue));
  }

  @Override
  public DrawerInfo<IntType> makeDrawerInfoInstance() {
    return new DrawerInfo<IntType>(IntTools.INSTANCE);
  }

  @Override
  public void setDrawerInfo(final int drawerIndex, final DrawerInfo<IntType> info) {
    assert drawerIndex >= 0;
    assert drawerIndex < this.numIndexedDrawers;

    this.current.drawerIndex = drawerIndex;

    // look up some things in the index
    // TODO (nbales) - Add support for value index.
    // int minValueLocation = this.firstEntryStart + (4 * drawerIndex);
    int drawerAddressLocation = this.drawerAddressStart + (4 * drawerIndex);

    int valuesOnDrawer = ENTIRES_IN_FULL_DRAWER;
    int firstEntry = drawerIndex * ENTIRES_IN_FULL_DRAWER;
    int drawerAddress = this.headerBuffer.getInt(drawerAddressLocation);

    info.reset(drawerIndex, drawerAddress, valuesOnDrawer, firstEntry);
  }

  @Override
  public boolean setDrawerInfoForEntryIndex(final int targetEntryIndex,
                                            final DrawerInfo<IntType> info) {
    long localIndex = targetEntryIndex - this.columnInfo.getFirstEntryIndex();

    // require this to be a valid index entry for this drawer
    assert localIndex >= 0 && localIndex < this.columnInfo.getEntryCount();

    int targetDrawer = (int) Math.floor(localIndex / ENTIRES_IN_FULL_DRAWER);
    this.setDrawerInfo(targetDrawer, info);
    return true;
  }

  @Override
  public boolean setDrawerInfoForValue(final IValueReference<IntType> value,
                                       final DrawerInfo<IntType> info) {
    int targetValue = this.intManip.getIntegerValue(value);
    int i = Math.max(0, this.current.drawerIndex);
    // faster with binary search
    for (; i < this.numIndexedDrawers; ++i) {
      int byteOffset = this.valueIndexStart + i * 4;
      int drawerMinValue = this.headerBuffer.getInt(byteOffset);
      if (drawerMinValue > targetValue) {
        int targetDrawer = i - 1;
        assert i >= 0;
        if (targetDrawer >= this.current.drawerIndex) {
          this.setDrawerInfo(targetDrawer, info);
          return true;
        }
      }
    }
    return false;
  }

  @Override
  public boolean setNextDrawerInfo(final DrawerInfo<IntType> info) {
    int nextDrawer = this.current.drawerIndex + 1;
    if (nextDrawer < this.numIndexedDrawers) {
      this.setDrawerInfo(nextDrawer, info);
      return true;
    } else {
      return false;
    }
  }

}
